package com.zz21

import com.gosun.{DataEncry, getResult}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.JavaConversions.asScalaBuffer
import scala.collection.mutable.ArrayBuffer

object zz21_vw_leftbehindpeoples {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession
      .builder()
      .config("spark.network.timeout", "1200")
      .master("local[*]")
      .appName("SQLContextApp")
      .getOrCreate()
    val options = Map(
      "es.nodes.wan.only" -> "true",
      "es.nodes" -> "10.82.121.72",
      "es.port" -> "9200",
      "es.read.field.as.array.include" -> "arr1, arr2",
      "es.scroll.size" -> "10000",
      "es.input.use.sliced.partitions" -> "false"
    )
    val index = "zz21_vw_leftbehindpeoples"
    val frame: DataFrame = sparkSession
      .read
      .format("es")
      .options(options)
      .load(index)
      .cache()
    //frame.show()

    val resSchema = StructType(
      List(
        StructField("HOUSEHOLDIDENTITY", StringType, true),
        StructField("ATTENTIONEXTENT", StringType, true),
        StructField("SOURCESSTATE", StringType, true),
        StructField("UPDATEDATE", StringType, true),
        StructField("CREATEDATE", StringType, true),
        StructField("UPDATEUSER", StringType, true),
        StructField("CREATEUSER", StringType, true),
        StructField("ISEMPHASISDATE", StringType, true),
        StructField("ISEMPHASISREASON", StringType, true),
        StructField("ISEMPHASIS", StringType, true),
        StructField("HELPINGSITUATION", StringType, true),
        StructField("DIFFICULTIESANDDEMANDS", StringType, true),
        StructField("FAMILYANNUALINCOME", StringType, true),
        StructField("FAMILYMAINADRESS", StringType, true),
        StructField("FAMILYMAINPHONE", StringType, true),
        StructField("LEFTRELATIONSHIP", StringType, true),
        StructField("FAMILYMAINHEALTH", StringType, true),
        StructField("FAMILYMAINNAME", StringType, true),
        StructField("FAMILYMAINIDCARD", StringType, true),
        StructField("LEFTTYPE", StringType, true),
        StructField("PERSONALANNUALINCOME", StringType, true),
        StructField("HEALTH", StringType, true),
        StructField("ORGINTERNALCODE", StringType, true),
        StructField("ORGID", StringType, true),
        StructField("ADDRESSINFOID", StringType, true),
        StructField("BASEINFOID", StringType, true),
        StructField("ID", StringType, true),
        StructField("FAMILYMAINIDCARD_AES", StringType, true)

      )
    )
    val value: RDD[Row] = frame.rdd.mapPartitions(iter => {
      val list = ArrayBuffer[Row]()
      while (iter.hasNext) {
        val row: Row = iter.next()
        val str1 = row.getAs[String]("HOUSEHOLDIDENTITY")
        val str2 = row.getAs[String]("ATTENTIONEXTENT")
        val str3 = row.getAs[String]("SOURCESSTATE")
        val str4 = row.getAs[String]("UPDATEDATE")
        val str5 = row.getAs[String]("CREATEDATE")
        val str6 = row.getAs[String]("UPDATEUSER")
        val str7 = row.getAs[String]("CREATEUSER")
        val str8 = row.getAs[String]("ISEMPHASISDATE")
        val str9 = row.getAs[String]("ISEMPHASISREASON")
        val str10 = row.getAs[String]("ISEMPHASIS")
        val str11 = row.getAs[String]("HELPINGSITUATION")
        val str12 = row.getAs[String]("DIFFICULTIESANDDEMANDS")
        val str13 = row.getAs[String]("FAMILYANNUALINCOME")
        val str14 = DataEncry.changAddress(row.getAs[String]("FAMILYMAINADRESS"))
        val str15 = row.getAs[String]("FAMILYMAINPHONE")
        val str16 = row.getAs[String]("LEFTRELATIONSHIP")
        val str17 = row.getAs[String]("FAMILYMAINHEALTH")
        val str18 = DataEncry.changName(row.getAs[String]("FAMILYMAINNAME"))
        val str19 = DataEncry.changIDcard(row.getAs[String]("FAMILYMAINIDCARD"))
        val str20 = row.getAs[String]("LEFTTYPE")
        val str21 = row.getAs[String]("PERSONALANNUALINCOME")
        val str22 = row.getAs[String]("HEALTH")
        val str23 = row.getAs[String]("ORGINTERNALCODE")
        val str24 = row.getAs[String]("ORGID")
        val str25 = row.getAs[String]("ADDRESSINFOID")
        val str26 = row.getAs[String]("BASEINFOID")
        val str27 = row.getAs[String]("ID")
        val str28 = DataEncry.changAES(row.getAs[String]("FAMILYMAINIDCARD"))
        val schema: GenericRowWithSchema = new GenericRowWithSchema(Array(
          str1,
          str2,
          str3,
          str4,
          str5,
          str6,
          str7,
          str8,
          str9,
          str10,
          str11,
          str12,
          str13,
          str14,
          str15,
          str16,
          str17,
          str18,
          str19,
          str20,
          str21,
          str22,
          str23,
          str24,
          str25,
          str26,
          str27,
          str28
        ), resSchema)
        list.append(schema)
      }
      list.iterator
    })
    val dataFrame: DataFrame = sparkSession.createDataFrame(value.coalesce(1).cache(), resSchema)
    val path = "C:\\Users\\Raichard\\Desktop\\宜春\\JSON\\zz21\\zz21_vw_leftbehindpeoples.json"

    dataFrame.show(false)
    val str: String = dataFrame.toJSON.collectAsList.mkString("[", ",", "]")
    getResult.getData(str, path)
    /*dataFrame
      .write
      .mode(SaveMode.Overwrite)
      .format("json")
      .save(path)*/
    println("输出完成！")
    sparkSession.close()

  }

}
